<?php
declare (strict_types=1);

namespace app\controller;

use app\BaseController;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\facade\Cache;

class Idempotent extends BaseController
{
    public function send()
    {
        // 建立连接
        $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'zq', '123456', '/');
        // 获取信道
        $channel = $connection->channel();

        // 创建DLX及死信队列
        $channel->exchange_declare('dlx_exchange', 'direct', false, false, false);
        $channel->queue_declare('dlx_queue', false, true, false, false);
        $channel->queue_bind('dlx_queue', 'dlx_exchange', 'dlx_routing_key');

        // 创建延迟队列
        $channel->exchange_declare('delay_exchange', 'direct', false, false, false);
        $args = new AMQPTable();
        // 消息过期方式：设置queue.normal队列中的消息5s之后过期
        $args->set('x-message-ttl', 5000);
        // 设置队列最大长度方式： x-max-length
//        $args->set('x-max-length', 1);
        $args->set('x-dead-letter-exchange', 'dlx_exchange');
        $args->set('x-dead-letter-routing-key', 'dlx_routing_key');
        $channel->queue_declare('delay_queue', false, true, false, false, false, $args);
        $channel->queue_bind('delay_queue', 'delay_exchange', 'delay_routing_key');

        // 模拟发送消息内容
        $messageBody = "重复消息发送（" . date("h:i:s") . "）";
        // 将我们需要的消息标记为持久化
        $message = new AMQPMessage($messageBody, [
            'content_type' => 'text/plain',
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
        ]);

        // 设置消息ID，防止重复消费
        $corr_id = uniqid();
        $headers = new AMQPTable([
            'correlation_id' => $corr_id
        ]);

        $message->set('application_headers', $headers);

        Cache::set($corr_id, $corr_id, 3600);

        // 绑定delay.routingKey路由
        $channel->basic_publish($message, 'delay_exchange', 'delay_routing_key');

        // 模拟发送重复消息
        $channel->basic_publish($message, 'delay_exchange', 'delay_routing_key');

        // 关闭信道
        $channel->close();
        // 关闭连接
        $connection->close();
        return 'send success';
    }
}
